package com.hzgj.bcl.soa.command;

import com.alibaba.fastjson.JSON;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.hzgj.bcl.stats.RedisKey;
import redis.clients.jedis.Tuple;

import java.util.*;
import java.util.concurrent.*;

/**
 * Created by Administrator on 2016/9/2.
 */
public class InvokeTimeCommand extends InvokeCommand {

    @Override
    String handleNode(String baseKey, long start, long end) {
        CountDownLatch latch = new CountDownLatch(4);
        Collection<? extends Callable<Object>> tasks = getTasks(baseKey, start, end, latch);
        List<Object> list = Lists.newArrayList();
        try {
            List<Future<Object>> futures = threadPool.invokeAll(tasks);
            latch.await(5, TimeUnit.SECONDS);
            futures.forEach(future -> {
                try {
                    list.add(future.get());
                } catch (Exception e) {
                    logger.error("{}", e);
                }
            });
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        return JSON.toJSONString(list);
    }

    private Collection<Callable<Object>> getTasks(String baseKey, long start, long end, CountDownLatch countDownLatch) {
        List<Callable<Object>> list = new ArrayList<>();
        list.add(() -> new CallableTemplate() {
            @Override
            Object invoke() {
                Map<String, String> map = Maps.newHashMap();
                Set<Tuple> tuples = redisClient.zrangeByScoreWithScores(RedisKey.getSpanTp75(baseKey), start, end);
                String result = buildTuple(tuples);
                map.put("name", "tp75");
                map.put("data", result);
                return map;
            }
        }.build(countDownLatch));
        list.add(() -> new CallableTemplate() {
            @Override
            Object invoke() {
                Map<String, String> map = Maps.newHashMap();
                Set<Tuple> tuples = redisClient.zrangeByScoreWithScores(RedisKey.getSpanTp95(baseKey), start, end);
                String result = buildTuple(tuples);
                map.put("name", "tp95");
                map.put("data", result);
                return map;
            }
        }.build(countDownLatch));
        list.add(() -> new CallableTemplate() {
            @Override
            Object invoke() {
                Map<String, String> map = Maps.newHashMap();
                Set<Tuple> tuples = redisClient.zrangeByScoreWithScores(RedisKey.getSpanTp99(baseKey), start, end);
                String result = buildTuple(tuples);
                map.put("name", "tp99");
                map.put("data", result);
                return map;
            }
        }.build(countDownLatch));
        list.add(() -> new CallableTemplate() {
            @Override
            Object invoke() {
                Map<String, String> map = Maps.newHashMap();
                Set<Tuple> tuples = redisClient.zrangeByScoreWithScores(RedisKey.getSpanTp999(baseKey), start, end);
                String result = buildTuple(tuples);
                map.put("name", "tp999");
                map.put("data", result);
                return map;
            }
        }.build(countDownLatch));
        return list;
    }

    abstract class CallableTemplate {
        public Object build(CountDownLatch countDownLatch) {
            try {
                return invoke();
            } catch (Exception e) {
                logger.error("{}", e);
            } finally {
                countDownLatch.countDown();
            }
            return "";
        }

        abstract Object invoke();

    }

    @Override
    public String getName() {
        return "invoke_time";
    }

    ThreadPoolExecutor threadPool = new ThreadPoolExecutor(4, 10, 10,
            TimeUnit.SECONDS, new ArrayBlockingQueue<>(10), new ThreadPoolExecutor.CallerRunsPolicy());
}
